package cn.cdqf.rabbitmq.task;

import cn.cdqf.rabbitmq.rabbit.RabbitSender;
import cn.cdqf.rabbitmq.util.RedisUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;

//@Component
@Slf4j
public class RabbitTask {
    @Autowired
    private RedisUtils redisUtils;
    @Autowired
    private RabbitSender rabbitSender;
    @Autowired
    private Executor getAsyncExecutor;
    @Scheduled(fixedDelay = 50*1000)
    public void repeatSend(){
        //通过redis的scan扫描所有rabbit_开头的key来进行判断
        Set<String> rabbitKeys = redisUtils.scan("rabbit_");
        rabbitKeys.parallelStream().forEach(key-> {
                    //分布式锁  第三个参数是 正常走业务的一个时间+5
                    if (redisUtils.hsetnx( key, "sendLock", String.valueOf(System.currentTimeMillis() + 2000 * 5))) {
                        try {
                            //多线程完成    如果断电  或者服务器崩溃
                            CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {
                                rabbitSender.sendMessageById( key);
                            }, getAsyncExecutor);
                            completableFuture.join();
                        } finally {

                            //释放锁
                            redisUtils.hdel( key, "sendLock");
                        }
                    } else {
                        Object object = redisUtils.hget( key, "sendLock");
                        //避免死锁,只能补偿  极端情况
                        if (object != null && Long.valueOf(String.valueOf(object)) < System.currentTimeMillis()) {
                            //在把时间改变一次
                            redisUtils.hset( key, "sendLock", String.valueOf(System.currentTimeMillis() + 2000 * 5));
                            //发送次数减一  多发一次而已
                            redisUtils.hincr( key, "sendCount", -1);
                            try {
                                //多线程完成
                                CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {
                                    rabbitSender.sendMessageById( key);
                                }, getAsyncExecutor);
                                completableFuture.join();
                            } finally {
                                //释放锁
                                redisUtils.hdel(key, "sendLock");
                            }
                        }
                    }
                }
        );
    }
}
